package com.whoami.idmagic

import org.apache.spark.sql.SparkSession

/**
 *
 * @author tzp
 * @since 2020/12/29
 */
object OOMTest {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("yarn").getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._
    import org.apache.spark.sql.functions._


    sc.setCheckpointDir("hdfs:///user/mz_supertool/workspace/tzp/20201229/checkpoint")
    val admDF = spark.read.orc("/user/mz_supertool/datalake/src/admonitor/ods_adm_bus/20201212/CAMPAIGN-m-00063.orc")
    val babelDF = spark.read.orc("/user/mz_supertool/datalake/src/babel/dim_adm_babel/orc/20201228")

    val admDF1 = admDF.select("raw_idfa", "mz_spot_id").filter($"raw_idfa".isNotNull)
//      .repartition($"mz_spot_id")
    val babelDF1 = babelDF.select("spots_id", "media_name")
//      .repartition($"spots_id")

    val mediaDF = admDF1.join(babelDF1, $"mz_spot_id" === $"spots_id", "left")
    mediaDF.write.orc("/user/mz_supertool/workspace/tzp/20201229/out3")
    mediaDF.persist()
    val mediaDF1 = mediaDF.checkpoint()


    val mediaDF2 = mediaDF.groupBy("media_name").agg(count("*").alias("media_log_count"))
    val mediaDF3 = mediaDF2.orderBy("media_log_count").take(100)

  }
}
